package com.example.springbootkafka.kafka;


import com.example.springbootkafka.constants.BaseConstants;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.streams.processor.*;
import org.springframework.util.SerializationUtils;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * 延时消息处理器
 *
 * @author huangchangjin
 * @date 2025/1/8 15:16
 */
@Slf4j
public class OldDelayProcessorSupplier implements ProcessorSupplier<String, String> {
    @Override
    public Processor<String, String> get() {
        return new Processor<String, String>() {
            private ProcessorContext context;

            @Override
            public void init(ProcessorContext context) {
                this.context = context;
            }


            @SneakyThrows
            @Override
            public void process(String key, String value) {
                Headers headers = context.headers();
                Long ttl = -1L;
                String deadTopic = "";
                // 模拟续约的情况
                for (Header header : headers) {
                    if (header.key().equals(BaseConstants.TTL_KEY)) {
                        ttl = (Long) SerializationUtils.deserialize(header.value());
                    }
                    if (header.key().equals(BaseConstants.DEAD_TOPIC_KEY)) {
                        deadTopic = (String) SerializationUtils.deserialize(header.value());
                    }
                }

                if(ttl == -1L){
                    return;
                }

//                context.schedule(Duration.ofMillis(ttl), PunctuationType.WALL_CLOCK_TIME, (timestamp) -> {
//                    log.info("timestamp = {} key = {}, value = {}, headers = {}", timestamp, key, value, headers);
//                    context.forward(key, value);
//                });

                TimeUnit.MILLISECONDS.sleep(ttl);
            }

            @Override
            public void close() {
                // 资源清理
            }
        };
    }
}
